package com.kipling.kio.core;

import java.nio.ByteOrder;
import java.nio.channels.SocketChannel;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.kipling.kio.core.PacketHandlerMode;
import com.kipling.kio.core.intf.KioHandler;
import com.kipling.kio.core.intf.Packet;
import com.kipling.kio.core.maintain.ChannelContextMapWithLock;
import com.kipling.kio.core.stat.GroupStat;
import com.kipling.kio.server.threadpool.DefaultThreadFactory;
import com.kipling.kio.server.threadpool.SynThreadPoolExecutor;
import com.kipling.kio.server.threadpool.intf.SynRunnableIntf;

public abstract class GroupContext< P extends Packet, R>
{

	public static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 1;

	public static final Semaphore SYN_SEND_SEMAPHORE = new Semaphore(1000000);

	/**
	 * 默认的心跳超时时间(单位: 毫秒)
	 */
	public static final long DEFAULT_HEARTBEAT_TIMEOUT = 1000 * 120;

	/** 
	 * 默认的接收数据的buffer size
	 */
	public static final int READ_BUFFER_SIZE = Integer.getInteger("tio.default.read.buffer.size", 2048);

	public static final long KEEP_ALIVE_TIME = 9000000L;

	private ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;

	/**
	 * 心跳超时时间(单位: 毫秒)
	 */
	protected long heartbeatTimeout = DEFAULT_HEARTBEAT_TIMEOUT;

	private PacketHandlerMode packetHandlerMode = PacketHandlerMode.SINGLE_THREAD;//.queue;

	protected ChannelContextMapWithLock< P, R> syns = new ChannelContextMapWithLock< P, R>();
	/**
	 * 接收数据的buffer size
	 */
	protected int readBufferSize = READ_BUFFER_SIZE;


	/**
	 * 低优先级的业务处理线程池
	 */
	private SynThreadPoolExecutor<SynRunnableIntf> handlerExecutorNormPrior = null;


	/**
	 * 低优先级的消息发送线程池
	 */
	private SynThreadPoolExecutor<SynRunnableIntf> sendExecutorNormPrior = null;

	/** The group executor. */
	protected ExecutorService groupExecutor = null;

	private ThreadPoolExecutor closePoolExecutor = null;

	/**
	 * packet编码成bytebuffer时，是否与ChannelContext相关，false: packet编码与ChannelContext无关
	 */
	private boolean isEncodeCareWithChannelContext = true;

	protected String id;

	private boolean isStopped = false;

	private final static AtomicInteger ID_ATOMIC = new AtomicInteger();

	public GroupContext()
	{
		super();
		this.id = ID_ATOMIC.incrementAndGet() + "";

		//		LinkedBlockingQueue<Runnable> poolQueueHighPrior = new LinkedBlockingQueue<Runnable>();
		//		SynThreadPoolExecutor<SynRunnableIntf> executorHighPrior = new SynThreadPoolExecutor<SynRunnableIntf>(CORE_POOL_SIZE, CORE_POOL_SIZE, KEEP_ALIVE_TIME, poolQueueHighPrior,
		//				DefaultThreadFactory.getInstance("t-aio-high-prior", Thread.MAX_PRIORITY), "t-aio-high-prior");
		//		executorHighPrior.prestartAllCoreThreads();

		LinkedBlockingQueue<Runnable> poolQueueNormPrior = new LinkedBlockingQueue<Runnable>();
		SynThreadPoolExecutor<SynRunnableIntf> executorNormPrior = new SynThreadPoolExecutor<SynRunnableIntf>(CORE_POOL_SIZE, CORE_POOL_SIZE, KEEP_ALIVE_TIME, poolQueueNormPrior,
				DefaultThreadFactory.getInstance("t-aio-norm-prior", Thread.NORM_PRIORITY), "t-aio-norm-prior");
		executorNormPrior.prestartAllCoreThreads();

		//		decodeExecutor = executorNormPrior;
		//		closeExecutor = executorNormPrior;//executorHighPrior;
		//		handlerExecutorHighPrior = executorNormPrior;//executorHighPrior;
		handlerExecutorNormPrior = executorNormPrior;
		//		sendExecutorHighPrior = executorNormPrior;//executorHighPrior;
		sendExecutorNormPrior = executorNormPrior;

		LinkedBlockingQueue<Runnable> closeQueue = new LinkedBlockingQueue<Runnable>();
		closePoolExecutor = new ThreadPoolExecutor(0, CORE_POOL_SIZE, 9, TimeUnit.SECONDS, closeQueue, DefaultThreadFactory.getInstance("t-aio-close", Thread.NORM_PRIORITY));

	}
	
	/**
     * 服务器端，保存userid和与客户端通信的socketChannel
     */
    public static Map<String, SocketChannel> socketChannelMap = new ConcurrentHashMap<String, SocketChannel>();
    
    /**
     * 返回服务器与客户端socketChannel
     * @param key
     * @return
     */
    public SocketChannel getSocketChannel(String key){
    	if(key ==null || "".equals(key))
    		return null;
    	
    	return socketChannelMap.get(key);
    }
    
    public void setSocketChannel(String key,SocketChannel socketChannel){
    	socketChannelMap.put(key, socketChannel);
    }

	/**
	 * 
	 * @return
	 * @author: tanyaowu
	 */
	public SynThreadPoolExecutor<SynRunnableIntf> getHandlerExecutorNormPrior()
	{
		return handlerExecutorNormPrior;
	}

	public static Map<String, SocketChannel> getSocketChannelMap() {
		return socketChannelMap;
	}

	public static void setSocketChannelMap(
			Map<String, SocketChannel> socketChannelMap) {
		GroupContext.socketChannelMap = socketChannelMap;
	}

	/**
	 * 
	 * @param handlerExecutorNormPrior
	 * @author: tanyaowu
	 */
	public void setHandlerExecutorNormPrior(SynThreadPoolExecutor<SynRunnableIntf> handlerExecutorNormPrior)
	{
		this.handlerExecutorNormPrior = handlerExecutorNormPrior;
	}

	/**
	 * 
	 * @return
	 * @author: tanyaowu
	 */
	public SynThreadPoolExecutor<SynRunnableIntf> getSendExecutorNormPrior()
	{
		return sendExecutorNormPrior;
	}

	/**
	 * 
	 * @param sendExecutorNormPrior
	 * @author: tanyaowu
	 */
	public void setSendExecutorNormPrior(SynThreadPoolExecutor<SynRunnableIntf> sendExecutorNormPrior)
	{
		this.sendExecutorNormPrior = sendExecutorNormPrior;
	}

	/**
	 * 
	 * @return
	 * @author: tanyaowu
	 */
	public ByteOrder getByteOrder()
	{
		return byteOrder;
	}

	/**
	 * 
	 * @param byteOrder
	 * @author: tanyaowu
	 */
	public void setByteOrder(ByteOrder byteOrder)
	{
		this.byteOrder = byteOrder;
	}

	/**
	 * 
	 * @return
	 * @author: tanyaowu
	 */
	public String getId()
	{
		return id;
	}

	/**
	 * @param id the id to set
	 */
	public void setId(String id)
	{
		this.id = id;
	}

	/**
	 * @return the heartbeatTimeout
	 */
	public long getHeartbeatTimeout()
	{
		return heartbeatTimeout;
	}

	/**
	 * @param heartbeatTimeout the heartbeatTimeout to set
	 */
	public void setHeartbeatTimeout(long heartbeatTimeout)
	{
		this.heartbeatTimeout = heartbeatTimeout;
	}


	/**
	 * @return the readBufferSize
	 */
	public int getReadBufferSize()
	{
		return readBufferSize;
	}

	/**
	 * @param readBufferSize the readBufferSize to set
	 */
	public void setReadBufferSize(int readBufferSize)
	{
		this.readBufferSize = readBufferSize;
	}

	/**
	 * @return
	 *
	 * @author: tanyaowu
	 * 2016年12月20日 上午11:32:02
	 * 
	 */
	public abstract KioHandler< P, R> getKioHandler();

	/**
	 * @return
	 *
	 * @author: tanyaowu
	 * 2016年12月20日 上午11:33:02
	 * 
	 */
	public abstract GroupStat getGroupStat();


	/**
	 * @param syns the syns to set
	 */
	public void setSyns(ChannelContextMapWithLock< P, R> syns)
	{
		this.syns = syns;
	}

	/**
	 * @return the isEncodeCareWithChannelContext
	 */
	public boolean isEncodeCareWithChannelContext()
	{
		return isEncodeCareWithChannelContext;
	}

	/**
	 * @param isEncodeCareWithChannelContext the isEncodeCareWithChannelContext to set
	 */
	public void setEncodeCareWithChannelContext(boolean isEncodeCareWithChannelContext)
	{
		this.isEncodeCareWithChannelContext = isEncodeCareWithChannelContext;
	}

	/**
	 * @return the isStop
	 */
	public boolean isStopped()
	{
		return isStopped;
	}

	/**
	 * @param isStop the isStop to set
	 */
	public void setStopped(boolean isStopped)
	{
		this.isStopped = isStopped;
	}

	/**
	 * @return the closePoolExecutor
	 */
	public ThreadPoolExecutor getClosePoolExecutor()
	{
		return closePoolExecutor;
	}

	/**
	 * @param closePoolExecutor the closePoolExecutor to set
	 */
	public void setClosePoolExecutor(ThreadPoolExecutor closePoolExecutor)
	{
		this.closePoolExecutor = closePoolExecutor;
	}

	/**
	 * @return the packetHandlerMode
	 */
	public PacketHandlerMode getPacketHandlerMode()
	{
		return packetHandlerMode;
	}

	/**
	 * @param packetHandlerMode the packetHandlerMode to set
	 */
	public void setPacketHandlerMode(PacketHandlerMode packetHandlerMode)
	{
		this.packetHandlerMode = packetHandlerMode;
	}

	/**
	 * @return the groupExecutor
	 */
	public ExecutorService getGroupExecutor()
	{
		return groupExecutor;
	}

	/**
	 * @param groupExecutor the groupExecutor to set
	 */
	public void setGroupExecutor(ExecutorService groupExecutor)
	{
		this.groupExecutor = groupExecutor;
	}

}
